package com.mfs.proxy.core.client.reactor;

import com.mfs.common.utills.ChannelUtil;
import com.mfs.common.utills.ObjectUtil;
import com.mfs.proxy.core.exceptions.ProxyException;
import com.mfs.proxy.core.client.config.ClientConfig;
import com.mfs.proxy.core.client.holder.ProxyServerHolder;
import com.mfs.proxy.core.enums.MessageTypeEnum;
import com.mfs.proxy.core.message.ChannelResponseMessageContent;
import com.mfs.proxy.core.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

public class SubClientReactor implements Runnable {
    private static final Logger logger = LoggerFactory.getLogger(SubClientReactor.class);

    private static Selector clientProxySelector;
    public SubClientReactor() {
        try {
            clientProxySelector = Selector.open();
        } catch (IOException e) {
            throw new ProxyException(e);
        }
    }

    public static SelectionKey register(SelectableChannel selectableChannel) throws ClosedChannelException {
        return selectableChannel.register(clientProxySelector.wakeup(), SelectionKey.OP_WRITE);
    }

    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                int select = clientProxySelector.select(100);
                if (select < 1) {
                    Thread.yield();
                    continue;
                }
                Set<SelectionKey> selectionKeys = clientProxySelector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    processReadableWritable(selectionKey);
                    iterator.remove();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 处理可读、可写事件
     * 可写：首次连接到服务端，回传请求编号
     * 可读：收到服务端传回的用户请求
     * */
    private void processReadableWritable(SelectionKey selectionKey) {
        if (selectionKey.isValid() && selectionKey.isWritable()) {
            Object attachment = selectionKey.attachment();
            if (attachment == null || !(attachment instanceof String)) {
                logger.error("错误的响应服务端通道请求");
                ChannelUtil.closeSocketChannel((SocketChannel)selectionKey.channel());
                selectionKey.cancel();
                return;
            }
            ChannelResponseMessageContent messageContent = new ChannelResponseMessageContent(ClientConfig.getInstance().getProxyHandle(), attachment.toString());
            try {
                SocketChannel proxyChanel = (SocketChannel)selectionKey.channel();
                ByteBuffer byteBuffer = ByteBuffer.wrap(ObjectUtil.serialize(new Message(MessageTypeEnum.CHANNEL_RESPONSE.getType(), messageContent)));
                proxyChanel.write(byteBuffer);
                selectionKey.interestOps(SelectionKey.OP_READ);
                //ProxyServerHolder.getProxyServer().getCommunicationMailbox().send(MessageTypeEnum.CHANNEL_RESPONSE, messageContent);
            } catch (IOException e) {
                logger.error("发送通道响应消息失败");
                ChannelUtil.closeSocketChannel((SocketChannel)selectionKey.channel());
                selectionKey.cancel();
                return;
            }
        } else if (selectionKey.isValid() && selectionKey.isReadable()) {
            // 取消注册，后续代理请求的处理到单独的线程中去
            selectionKey.cancel();
            ProxyServerHolder.getProxyServer().processRequest((SocketChannel) selectionKey.channel());
        }
    }

}
